package cn.gwm.flink.streaming.dwd.streamsource;

import cn.hutool.json.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.experimental.Accessors;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @ClassName FilterStreamSource
 * @Description
 * @Author LiangGuang
 * @Date 2023/06/15 15:19
 */
@Data
public abstract class FilterStreamSource extends StreamSource {

    public CommonStreamSource source;

    public FilterStreamSource(CommonStreamSource source) {
        this.source = source;
    }

    @Override
    public DataStream<JSONObject> source(StreamExecutionEnvironment env, String groupId, String topic) {
        DataStream<JSONObject> dataStream = source.source(env, groupId, topic);
        return filter(dataStream);
    }

    public abstract DataStream<JSONObject> filter(DataStream<JSONObject> dataStream);
}
